package com.cande;

import io.reactivex.rxjava3.core.Flowable;
import io.reactivex.rxjava3.core.Observable;
import io.reactivex.rxjava3.core.ObservableOnSubscribe;
import io.reactivex.rxjava3.core.Observer;
import io.reactivex.rxjava3.disposables.Disposable;
import org.reactivestreams.Subscriber;
import org.reactivestreams.Subscription;

import java.nio.file.Paths;
import java.util.LinkedHashMap;

public class Main {
    public static void main(String[] args) {
        System.out.println(Paths.get("").toAbsolutePath());

        Flowable.just("Hello world").subscribe(System.out::println);
        //observer 无背压
//        observer();

        //flowable 有背压
        flowable();
    }

    public static void observer(){
        Observable mObservable = Observable.create((ObservableOnSubscribe<Integer>) e -> {
            e.onNext(1);
            e.onNext(2);
            e.onComplete();
        });

        Observer mObserver=new Observer<Integer>() {
            //这是新加入的方法，在订阅后发送数据之前，
            //回首先调用这个方法，而Disposable可用于取消订阅
            @Override
            public void onSubscribe(Disposable d) {

            }

            @Override
            public void onNext(Integer value) {
                System.out.println("lucas ，onNext: "+value );
            }

            @Override
            public void onError(Throwable e) {

            }

            @Override
            public void onComplete() {

            }
        };

        mObservable.subscribe(mObserver);

    }


    public static void flowable(){
        Flowable.range(1,10).subscribe(new Subscriber<Integer>() {
                                           Subscription sub;
                                           //当订阅后，会首先调用这个方法，其实就相当于onStart()，
                                           //传入的Subscription s参数可以用于请求数据或者取消订阅
                                           @Override
                                           public void onSubscribe(Subscription s) {
                                               System.out.println("TAG,onsubscribe start");
                                               sub=s;
                                               sub.request(1);
                                               System.out.println("TAG,onsubscribe end");
                                           }

                                           @Override
                                           public void onNext(Integer o) {
                                               System.out.println("TAGonNext--->"+o);
                                               sub.request(1);
                                           }
                                           @Override
                                           public void onError(Throwable t) {
                                               t.printStackTrace();
                                           }
                                           @Override
                                           public void onComplete() {
                                               System.out.println("TAG ,onComplete");
                                           }
                                       });
    }
}